package com.doit.doit47;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.Properties;

public class _09_生产者拦截器和分区器 {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux01:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //拦截器
        props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyInterceptor.class.getName());

//        props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());
        props.setProperty("retries",">0");
        props.setProperty("max.in.flight.requests.per.connection","<=5");

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        EventLog eventLog = new EventLog();


        for (int i = 0; i < 100; i++) {
            /*eventLog.setGuid(RandomUtils.nextInt(1, 10000000));
            eventLog.setEventId(RandomStringUtils.randomAlphabetic(10));
            eventLog.setTimestamp(System.currentTimeMillis());
            String json = JSON.toJSONString(eventLog);
            if (i %3 == 0){
                json += "}";
            }
            System.out.println(json);*/
            producer.send(new ProducerRecord<String, String>("test", i+""));

//            Thread.sleep(200);
        }
//        producer.flush();
    }
}
